package com.hyl.msg;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

public class RockConsumer {

	private String _groupName = "cn.js.nt.hyl";
	private String _namesrvAddr = "192.168.0.138:9876";
	private DefaultMQPushConsumer _consumer = null;

	public static RockConsumer getInstance(String groupName, String serverurl) {
		if (groupName == null || serverurl == null)
			return null;

		return new RockConsumer(groupName, serverurl);
	}

	private RockConsumer(String groupName, String serverurl) {
		/**         
		 *  一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>   		 
		* 注意：ProducerGroupName需要由应用来保证唯一<br>         
		*  ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，         
		*   因为服务器会回查这个Group下的任意一个Producer         */
		_groupName = groupName;
		_namesrvAddr = serverurl;
		//需要一个consumer group名字作为构造方法的参数，这里为consumer1 
		_consumer = new DefaultMQPushConsumer(_groupName);
		//同样也要设置NameServer地址 
		_consumer.setNamesrvAddr(_namesrvAddr);
		//Consumer后启动，所以MQ上也就堆积了一堆数据，Consumer的
		_consumer.setConsumeMessageBatchMaxSize(10);//每次拉取10条    

	}
	/**接收广播*/
	public boolean pullBroadMsg(String topic, String tags, MessageListenerConcurrently listener)
			throws MQClientException {
		_consumer.setMessageModel(MessageModel.BROADCASTING);
		return pullMsg(topic,tags,listener);
	}
	public boolean pullMsg(String topic, String tags, MessageListenerConcurrently listener)
			throws MQClientException {

		//这里设置的是一个consumer的消费策略 
		//CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息 
		//CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍 
		//CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前 
		_consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		//设置consumer所订阅的Topic和Tag，*代表全部的Tag 
		if (topic == null || topic.isEmpty())
			return false;
		if (tags == null || tags.isEmpty())
			tags = "*";
		
		_consumer.subscribe(topic, tags);
		//设置一个Listener，主要进行消息的逻辑处理 
		if (listener == null)
			listener = new MessageListenerConcurrently() {
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
						ConsumeConcurrentlyContext context) {
					//System.out.println(Thread.currentThread().getName() + " 接收消息: " + msgs );
					//返回消费状态
					//CONSUME_SUCCESS 消费成功 
					//RECONSUME_LATER 消费失败，需要稍后重新消费 
					try {
						// 06接收消息并打印
						for (MessageExt msg : msgs) {
							String topic = msg.getTopic();
							String msgBody = new String(msg.getBody(), "utf-8");
							String tags = msg.getTags();
							System.out.println("收到消息： topic:" + topic + " ,tags:" + tags + " ,msg: " + msgBody);
						}
					} catch (Exception e) {
						e.printStackTrace();
						// 1s 2s 5s ... 2h
						   return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试  
					}
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
			};
		_consumer.setMessageListener(listener);
		//调用start()方法启动consumer
		_consumer.start();
		return true;

	}

	public void close() {
		_consumer.shutdown();
	}
}
